package otSpaces.msg;

import java.nio.ByteBuffer;

import otSpaces.OtSpaceFunctions;
import otSpaces.util.OrdinalBytes;

public class Message {

	public static class UnknownMessageTypeException extends Exception {

		private static final long serialVersionUID = 8892410904393077888L;

		public UnknownMessageTypeException(final byte type) {
			super("Recieved a message of unknown type [" + Byte.toString(type)
					+ "]");
		}

	}

	public static final int MAX_BUFFER = 1024;

	public static final OrdinalBytes TYPE_BYTES = new OrdinalBytes(
			MessageTypes.values());

	/**
	 * Reads a buffer into a message object
	 */
	public static Message create(ByteBuffer buffer)
			throws UnknownMessageTypeException {

		final byte typeB = buffer.get();
		final short id = buffer.getShort();

		switch (MessageTypes.valueOf(TYPE_BYTES.valueOf(typeB).name())) {
		case JOIN:
			return createJoin(id, getString(getBuffer(buffer)));

		case LEAVE:
			return createLeave(id);

		case DOWNSTREAM:
			return createDownstream(id, toByteArray(getBuffer(buffer)));

		case UPSTREAM: {

			final byte folded = buffer.get();
			final ByteBuffer contents = getBuffer(buffer);

			return createUpstream(id, folded, toByteArray(contents));
		}

		default:
			throw new UnknownMessageTypeException(typeB);
		}
	}

	public static Message createDownstream(short topicId, byte[] data) {
		return new Message(MessageTypes.DOWNSTREAM, topicId, null, null, data);
	}

	public static Message createJoin(final short topicId, final String topicName) {
		return new Message(MessageTypes.JOIN, topicId, topicName, null, null);
	}

	public static Message createLeave(short topicId) {
		return new Message(MessageTypes.LEAVE, topicId, null, null, null);
	}

	public static Message createUpstream(short _id, byte foldHashCode,
			byte[] byteArray) {
		return new Message(MessageTypes.UPSTREAM, _id, null, foldHashCode,
				byteArray);
	}

	/**
	 * Reads a sub-buffer out of a buffer. Modifies the Buffer as if it was
	 * reading a prim
	 */
	public static ByteBuffer getBuffer(ByteBuffer container) {

		final int length = (container.getShort()) - (Short.MIN_VALUE);
		{
			final int remaining = container.remaining();
			assert remaining >= length;
		}

		final ByteBuffer ret = ByteBuffer.allocate(length);

		while (ret.hasRemaining()) {
			ret.put(container.get());
		}

		assert ret.position() == length;

		ret.flip();

		assert ret.limit() == length;

		return ret;
	}

	/**
	 * Reads a string out of a buffer. Modifies the Buffer as if it was reading
	 * a prim
	 */
	public static String getString(ByteBuffer buffer) {
		final int length = buffer.getShort() - Short.MIN_VALUE;

		final StringBuilder builder = new StringBuilder();

		while (builder.length() < length) {
			builder.append(buffer.getChar());
		}

		return builder.toString();
	}

	/**
	 * Puts a inner / sub buffer into an outer buffer. Modifies the outer Buffer
	 * as if it was writing a prim, but tyhe inner buffer's position should be
	 * restored to how it was at teh start of the call (and mark is unaffected)
	 */
	public static void putBuffer(ByteBuffer outer, ByteBuffer inner) {

		final int position = inner.position();

		assert (inner.remaining() + Short.MIN_VALUE) < Short.MAX_VALUE;

		// add the buffer's size
		outer.putShort((short) (inner.limit() + Short.MIN_VALUE));

		// add the buffer's data
		while (inner.hasRemaining()) {
			outer.put(inner.get());
		}

		inner.position(position);
	}

	/**
	 * Puts a String into a buffer. Modifies the outer Buffer as if it was
	 * writing a prim
	 */
	public static void putString(ByteBuffer buffer, String string) {
		final int encodedLength = (string.length() + Short.MIN_VALUE);
		assert encodedLength < Short.MAX_VALUE;

		buffer.putShort((short) encodedLength);

		for (char c : string.toCharArray()) {
			buffer.putChar(c);
		}
	}

	/**
	 * Turns a buffer into a byte array. The buffer's position should be
	 * restored to how it was at teh start of the call (and mark is unaffected)
	 * 
	 * @param data
	 * @return
	 */
	public static byte[] toByteArray(ByteBuffer data) {
		final byte[] ret = new byte[data.remaining()];

		final int position = data.position();

		data.get(ret);

		data.position(position);

		return ret;
	}

	public final byte[] _contents;

	public final Byte _foldedTopicNameHashCode;

	public final short _topicId;

	public final String _topicName;

	public final MessageTypes _type;

	/**
	 * DON'T USE THIS DIRECTLY
	 */
	private Message(final MessageTypes type, final short topicId,
			final String topicName, final Byte foldedTopicNameHashCode,
			final byte[] contents) {

		assert topicName == null
				|| foldedTopicNameHashCode == null
				|| OtSpaceFunctions.foldHashCode(topicName) == foldedTopicNameHashCode
						.byteValue();

		_type = type;
		_topicId = topicId;
		_topicName = topicName;
		_foldedTopicNameHashCode = foldedTopicNameHashCode;
		_contents = contents;
	}

	public ByteBuffer toBuffer() {
		final ByteBuffer ret = ByteBuffer.allocate(MAX_BUFFER);

		// put type
		ret.put(OrdinalBytes.ordinalByte(_type));

		// put topicId
		ret.putShort(_topicId);

		// start the contents buffer
		{
			final ByteBuffer contents = ByteBuffer.allocate(MAX_BUFFER - 5);

			// put topic name (iff we 'should')
			if (_topicName != null) {

				// only join should have this
				assert _type == MessageTypes.JOIN;

				putString(contents, _topicName);
			}

			// put foldedTopicNameHashCode (iff we 'should')
			if (_foldedTopicNameHashCode != null) {

				// only upstream or leav should have this
				assert _type == MessageTypes.UPSTREAM;

				ret.put(_foldedTopicNameHashCode.byteValue());
			}

			// put any other data (iff we 'should')
			if (_contents != null) {

				// only upstream and downstream should have this
				assert _type == MessageTypes.UPSTREAM
						|| _type == MessageTypes.DOWNSTREAM;

				putBuffer(ret, ByteBuffer.wrap(_contents));
			}

			// flip
			contents.flip();

			assert (contents.limit() + Short.MIN_VALUE) < Short.MAX_VALUE;

			// add the contents buffer
			putBuffer(ret, contents);
		}

		// return the whole damn thing
		return (ByteBuffer) ret.flip();
	}
}
